
package org.apache.ocean.util;

import java.util.concurrent.*;
import java.util.concurrent.locks.*;
import java.util.*;
import org.apache.ocean.*;

/**
 *
 * @author jasonr
 */
public class CircularQueue<T> {
  private LinkedBlockingQueue<T> queue;
  private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
  
  public CircularQueue() {
    queue = new LinkedBlockingQueue<T>();
  }
  
  public CircularQueue(int size) {
    queue = new LinkedBlockingQueue<T>(size);
  }
  
  public CircularQueue(Collection<T> collection) {
    queue = new LinkedBlockingQueue<T>();
    set(collection);
  }
  
  public Collection getCollection() {
    return queue;
  }
  
  public String toString() {
    return queue.toString();
  }
  
  public void removeAll() {
    LockUtil.lock(lock.writeLock());
    try {
      queue.clear();
    } finally {
      lock.writeLock().unlock();
    }
  }
  
  public void set(Collection<T> collection) {
    LockUtil.lock(lock.writeLock());
    try {
      queue.clear();
      queue.addAll(collection);
    } finally {
      lock.writeLock().unlock();
    }
  }
  
  public void add(T object) throws InterruptedException {
    LockUtil.lock(lock.readLock());
    try {
      queue.put(object);
    } finally {
      lock.readLock().unlock();
    }
  }
  
  public boolean remove(T object) {
    LockUtil.lock(lock.readLock());
    try {
      return queue.remove(object);
    } finally {
      lock.readLock().unlock();
    }
  }
  
  public int size() {
    LockUtil.lock(lock.readLock());
    try {
      return queue.size();
    } finally {
      lock.readLock().unlock();
    }
  }
  
  public T poll(long time, TimeUnit timeUnit) throws InterruptedException {
    LockUtil.lock(lock.readLock());
    try {
      T object = queue.poll(time, timeUnit);
      if (object != null) queue.add(object);
      return object;
    } finally {
      lock.readLock().unlock();
    }
  }
  
  public T peek() {
    LockUtil.lock(lock.readLock());
    try {
      return queue.peek();
    } finally {
      lock.readLock().unlock();
    }
  }
  
  public T poll() {
    LockUtil.lock(lock.readLock());
    try {
      T object = queue.poll();
      if (object != null) queue.add(object);
      return object;
    } finally {
      lock.readLock().unlock();
    }
  }
}
